unit uCxtClientNet;

interface

uses
  SysUtils, Classes, Windows, uDZStream, uDZPersistent, uDZWinsock2, uDZSocketUtils,
  uDZSocket, uDZThread, uDZTrace, uDZCRC, uDZBinaryTree, uCxtConst, uCxtTypes,
  uCxtPackage, uAppConfig, uAppLog, uGUDPSpeedTest;

type
  TPkgProcessor = procedure(Request: TRequestPackage; Response: TResponsePackage);
  TRequestErrorCallback = procedure(Request: TRequestPackage);

procedure AddRequest(Package: TRequestPackage);

var
  TcpClient: TDzClientSocket;
  PkgProcessor: TPkgProcessor;
  OnConnectFail: TRequestErrorCallback;
  OnSendError: TRequestErrorCallback;
  OnRequestAbort: TRequestErrorCallback;
  
  UseCustomServer: Boolean;
  CustomServerIP: AnsiString;
  CustomServerPort: Integer;

  ServerEntries: PSocketSvcEntry;
  ServerIPs: array of AnsiString;

implementation

var
  SendStream: TDzMemoryStream;
  SendSocketEvent: THandle;
  RecvStream: TDzMemoryStream;
  RecvSocketEvent: THandle;

type
  TSendThread = class(TDzThread)
  private
    function CheckConnection: Boolean;
    function ExtractQueue(Timeout: LongWord = INFINITE): TRequestPackage;
    function SendPackage(Package: TRequestPackage): Boolean;
  protected
    procedure Execute; override;
  end;

  TRecvThread = class(TDzThread)
  private
    PkgHdr: TPackageHeader;
    function Read(var Buf; BufLen: Integer): Boolean;
  protected
    procedure Execute; override;
  end;
  
var
  PkgIdentity: Integer;
  PkgSem: THandle;  
  QueueLock: TDzCriticalSection;
  QueuePkgs: TRequestPackage;
  SentLock: TDzCriticalSection;
  SentPkgs: TRequestPackage;
  SendThread: TSendThread;
  RecvThread: TRecvThread;

procedure ClearPkgLinkList(Lock: TDzCriticalSection; var List: TRequestPackage);
var
  Pkg: TRequestPackage;
begin
  Lock.Acquire;
  try
    while Assigned(List) do
    begin
      Pkg := List;
      List := List.ForwardLink;
      Pkg.Free;
    end;
  finally
    Lock.Release;
  end;
end;

procedure IteratePkgLinkList(Lock: TDzCriticalSection; var List: TRequestPackage;
  Callback: TRequestErrorCallback);
var
  Pkg: TRequestPackage;
begin
  Lock.Acquire;
  try
    while Assigned(List) do
    begin
      Pkg := List;
      List := List.ForwardLink;
      Callback(Pkg);
    end;
  finally
    Lock.Release;
  end;
end;

procedure AddRequest(Package: TRequestPackage);
begin
  QueueLock.Acquire;
  try
    if Assigned(QueuePkgs) and (QueuePkgs.ReqType = rtLogin) then
    begin
      Package.ForwardLink := QueuePkgs.ForwardLink;
      QueuePkgs.ForwardLink := Package;
    end
    else begin
      Package.ForwardLink := QueuePkgs;
      QueuePkgs := Package;
    end;
    ReleaseSemaphore(PkgSem, 1, nil);
  finally
    QueueLock.Release;
  end;    
end;

procedure AddToSent(Package: TRequestPackage);
begin
  SentLock.Acquire;
  try
    Package.ForwardLink := SentPkgs;
    SentPkgs := Package;
  finally
    SentLock.Release;
  end;
end;

function ExtractSent(SerialNo: Integer): TRequestPackage;
var
  Backward, Pkg: TRequestPackage;
begin
  Backward := nil;
  Pkg := SentPkgs;
  Result := nil;
  SentLock.Acquire;
  try
    while Assigned(Pkg) do
    begin
      if Pkg.SerialNo = SerialNo then
      begin
        Result := Pkg;
        if Assigned(Backward) then Backward.ForwardLink := Pkg.ForwardLink
        else SentPkgs := Pkg.ForwardLink;
        Break;
      end
      else begin
        Backward := Pkg;
        Pkg := Pkg.ForwardLink;
      end;
    end;
  finally
    SentLock.Release;
  end;
end;

procedure RemoveEntry(const IP: AnsiString; Port: Word);
var
  pBackward, pEntry: PSocketSvcEntry;
begin
  pBackward := nil;
  pEntry := ServerEntries;
  while Assigned(pEntry) do
  begin                                                         
    if pEntry^.IP = IP then
    begin
      pEntry^.DeletePort(Port);
      if pEntry^.PortCount = 0 then
      begin
        if pBackward = nil then ServerEntries := pEntry^.ForwardLink
        else pBackward^.ForwardLink :=  pEntry^.ForwardLink;
        Dispose(pEntry);
      end;
      Break;
    end
    else begin
      pBackward := pEntry;
      pEntry := pEntry^.ForwardLink;
    end;
  end;  
end;

procedure DisposeServerEntries;
var
  pEntry: pSocketSvcEntry;
begin
  while Assigned(ServerEntries) do
  begin
    pEntry := ServerEntries;
    ServerEntries := ServerEntries^.ForwardLink;
    Dispose(pEntry);
  end;
end;

procedure DoSpeedTest;
begin
  try
    if Length(ServerIPs) = 0 then ServerEntries := SpeedTest(ServerIPs)
    else ServerEntries := SpeedTest(ServerIPArray);
  except

  end;

  if not Assigned(ServerEntries) then
  begin
    New(ServerEntries);
    ServerEntries^.IP := '60.172.42.46';
    ServerEntries^.Name := 'default cxt server';
    ServerEntries^.Delay := 0;
    ServerEntries^.AddPort(7890);
    ServerEntries^.ForwardLink := nil;  
  end;
end;

procedure PrepareRequestStream(Package: TRequestPackage);
var
  Header: TPackageHeader;
  Bookmark: Integer;
begin
  Header.Fixed.Signature := 'cxtc';
  Header.Fixed.HdrLen := SizeOf(Header);
  Header.Fixed.ObjFmt := Ord(afBinary);
  Header.Fixed.ReqType := Ord(Package.ReqType);
  SendStream.Seek(SizeOf(Header), soFromBeginning);
  Package.SaveToStream(SendStream, afBinary);
  Bookmark := SendStream.Position;
  Header.Fixed.DataLen := Bookmark - SizeOf(Header);
  Header.CheckSum := CRC32(@Header, Header.Fixed.HdrLen - SizeOf(Header.CheckSum));
  SendStream.Seek(0, soFromBeginning);
  SendStream.Write(Header, Header.Fixed.HdrLen);
  SendStream.Position := Bookmark;
end;

{ TRecvThread }

procedure TRecvThread.Execute;
var
  VarHdrLen: Integer;
  Response: TResponsePackage;
begin
  inherited;
  while not Stopped do
  begin
    if not TcpClient.Connected then Suspend;
    if not Read(PkgHdr, SizeOf(TFixedPackageHeader)) then Continue;
    if not PkgHdr.Fixed.IsValidResponse then
    begin
      WriteLog(tsWarning, 'invalid package format'#10);
      TcpClient.Close;
      IteratePkgLinkList(SentLock, SentPkgs, OnRequestAbort);
      Continue;
    end;
    VarHdrLen := PkgHdr.Fixed.HdrLen - SizeOf(TFixedPackageHeader);
    if (VarHdrLen > 0) and not Read(Pointer(Integer(@PkgHdr) +
      SizeOf(TFixedPackageHeader))^, VarHdrLen) then Continue;
    if RecvStream.Size < PkgHdr.Fixed.DataLen then
      RecvStream.Size := PkgHdr.Fixed.DataLen;
    if not Read(RecvStream.Memory^, PkgHdr.Fixed.DataLen) then Continue;
    Response := ResponseClasses[TRequestType(PkgHdr.Fixed.ReqType)].Create;
    RecvStream.Position := 0;
    Response.LoadFromStream(RecvStream, TDzArchiveFormat(PkgHdr.Fixed.ObjFmt));
    Writeln(Response.AsXML);
    PkgProcessor(ExtractSent(Response.SerialNo), Response);
  end;
end;

function TRecvThread.Read(var Buf; BufLen: Integer): Boolean;
var
  BytesRead, Flags: DWORD;
  Overlap: TWsaOverlapped;
  TtlRead: Integer;
begin
  Result := False;
  TtlRead := 0;
  while TtlRead < BufLen do
  begin
    FillChar(Overlap, SizeOf(Overlap), 0);
    Overlap.hEvent := RecvSocketEvent;
    WSAResetEvent(RecvSocketEvent);
    try
      TcpClient.AsyncRecv(Pointer(Integer(@Buf) + TtlRead)^, BufLen - TtlRead,
        @Overlap, nil, nil, 0);
    except
      on E: Exception do
      begin
        TcpClient.Close;
        IteratePkgLinkList(SentLock, SentPkgs, OnRequestAbort);
        LogException(E);
        Exit;
      end;
    end;
    if WAIT_OBJECT_0 <> WaitForMultipleObjects(1, @RecvSocketEvent, INFINITE) then Exit;

    if not WSAGetOverlappedResult(TcpClient.Handle, @Overlap,
      BytesRead, False, Flags) then Exit;

    Inc(TtlRead, BytesRead);

  end;

  Result := True;
end;

{ TSendThread }

function TSendThread.CheckConnection: Boolean;
var
  pEntry: PSocketSvcEntry;
  Overlap: TWsaOverlapped;
  Bytes, Flags: DWORD;
begin
  if TcpClient.Connected then
  begin
    Result := True;
    Exit;
  end;

  if UseCustomServer then
  begin
    DisposeServerEntries;
    New(ServerEntries);
    ServerEntries^.IP := CustomServerIP;
    ServerEntries^.AddPort(CustomServerPort);
    ServerEntries^.Delay := 0;
    ServerEntries^.Name := 'user defined';
    ServerEntries^.ForwardLink := nil;
  end
  else if not Assigned(ServerEntries) then DoSpeedTest;

  while Assigned(ServerEntries) do
  begin
    while ServerEntries^.PortCount > 0 do
    begin
      TcpClient.AllocateHandle(AF_INET, SOCK_STREAM, nil, True);
      FillChar(Overlap, SizeOf(Overlap), 0);
      Overlap.hEvent := SendSocketEvent;
      WSAResetEvent(SendSocketEvent);
      try
        TcpClient.AsyncConnect(ServerEntries^.IP, ServerEntries^.Ports[0],
          @Overlap, nil, nil);
        if (WAIT_OBJECT_0 = Self.WaitForMultipleObjects(1, @SendSocketEvent)) and
          WSAGetOverlappedResult(TcpClient.Handle, @Overlap, Bytes, False, Flags) then
        begin
          AppConfig.ServerIP := ServerEntries^.IP;
          AppConfig.ServerPort := ServerEntries^.Ports[0];
          TcpClient.UpdateConnectContext;
          RecvThread.Resume(True);
          Result := True;
          Exit;
        end;
      except
      end;
      ServerEntries^.Delete(0);
      if ServerEntries^.PortCount = 0 then
      begin
        pEntry := ServerEntries;
        ServerEntries := ServerEntries^.ForwardLink;
        Dispose(pEntry);
        Break;
      end;
    end;
  end;

  Result := False;
end;

procedure TSendThread.Execute;
var
  Request: TRequestPackage;
  SerialNo: Integer;
  ReqType: TRequestType;
begin
  inherited;
  while not Stopped do
  begin
    Request := ExtractQueue(3 * 60 * 1000);
    if not Assigned(Request) then
    begin
      if not Assigned(SentPkgs) then TcpClient.Close;
      Continue;
    end;
    if not CheckConnection then
    begin
      OnConnectFail(Request);
      Continue;
    end;
    ReqType := Request.ReqType;
    SerialNo := InterlockedIncrement(PkgIdentity);
    Request.SerialNo := SerialNo;
    if Assigned(ResponseClasses[ReqType]) then AddToSent(Request);
    if not SendPackage(Request) then
    begin
      if Assigned(ResponseClasses[ReqType]) then Request := ExtractSent(SerialNo);
      if Assigned(Request) then
      begin
        OnSendError(Request);
        AddRequest(Request);
      end;
    end
    else if not Assigned(ResponseClasses[ReqType]) then Request.Free;
  end;
end;

function TSendThread.ExtractQueue(Timeout: LongWord): TRequestPackage;
begin
  if Self.WaitForMultipleObjects(1, @PkgSem, Timeout) = WAIT_OBJECT_0 then
  begin
    QueueLock.Acquire;
    try
      Result := QueuePkgs;
      if Assigned(QueuePkgs) then QueuePkgs := QueuePkgs.ForwardLink;
    finally
      QueueLock.Release;
    end;
  end
  else Result := nil;
end;

function TSendThread.SendPackage(Package: TRequestPackage): Boolean;
var
  Overlap: TWsaOverlapped;
  BytesSent, Flags: DWORD;
begin
  Result := False;
  PrepareRequestStream(Package);
  FillChar(Overlap, SizeOf(Overlap), 0);
  Overlap.hEvent := SendSocketEvent;
  WSAResetEvent(SendSocketEvent);
  try
    TcpClient.AsyncSend(SendStream.Memory^, SendStream.Position, @Overlap, nil, nil, 0);
  except 
    on E: Exception do 
    begin 
      TcpClient.Close;
      IteratePkgLinkList(SentLock, SentPkgs, OnRequestAbort);
      LogException(E);
      Exit;
    end;
  end;
  
  if WAIT_OBJECT_0 <> WaitForMultipleObjects(1, @SendSocketEvent, INFINITE) then Exit;

  Result := WSAGetOverlappedResult(TcpClient.Handle, @Overlap, BytesSent, False, Flags);
    
  if Result then FmtWriteLog(tsDebug, 'send %s request(%d bytes)'#10,
    [ReqTypeNames[Package.ReqType],
    PPackageHeader(SendStream.Memory)^.Fixed.DataLen]);
end;

procedure CleanupThread(Thread: TDzThread);
begin
  Thread.Stop;
  Thread.Resume(True);
  Thread.WaitForTerminate;
  Thread.ReleaseThread;
  Thread.Free;
end;

procedure SendLogout;
var
  Any: Integer;
  Request: TREQLogout;
begin
  if not TcpClient.Connected then Exit;
  Any := 0;
  Request := TREQLogout.Create;
  try
    PrepareRequestStream(Request);
    try
      TcpClient.Write(SendStream.Memory^, SendStream.Position);
      WriteLog(tsDebug, 'logout'#10);
      TcpClient.Read(Any, SizeOf(Any));
    except
      on E: Exception do
      begin
        TcpClient.Close;
        LogException(E);
      end;
    end;
  finally
    Request.Free;
  end;
end; 

procedure Initialize;
const
  _100KB = 1024 * 100;
  _8KB = 8192;
begin
  SendStream := TDzMemoryStream.Create;
  SendStream.Size := _8KB;
  SendSocketEvent := WSACreateEvent;
  RecvStream := TDzMemoryStream.Create;
  RecvStream.Size := _100KB;
  RecvSocketEvent := WSACreateEvent;
  PkgSem := CreateSemaphore(nil, 0, MaxInt, nil);
  QueueLock := TDzCriticalSection.Create(100);
  SentLock := TDzCriticalSection.Create(100);
  TcpClient := TDzClientSocket.Create;
  SendThread := TSendThread.Create(nil, [tdUseCOM, tdStopSignal]);
  SendThread.Start;
  RecvThread := TRecvThread.Create(nil, [tdUseCOM, tdStopSignal]);
  RecvThread.Start;
end;

procedure Finalize;
begin
  CleanupThread(SendThread);
  CleanupThread(RecvThread);
  SendLogout;
  TcpClient.Free;                              
  CloseHandle(PkgSem);
  DisposeServerEntries;
  ClearPkgLinkList(SentLock, SentPkgs);
  ClearPkgLinkList(QueueLock, QueuePkgs);
  QueueLock.Free;
  SentLock.Free;
  SendStream.Free;
  WSACloseEvent(SendSocketEvent);
  RecvStream.Free;
  WSACloseEvent(RecvSocketEvent);
end;

initialization
  Initialize;

finalization
  Finalize;
  
end.
